package nl.dxn.tunnel.client.handel;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.util.internal.EmptyArrays;
import nl.dxn.tunnel.CollectionUtils;
import nl.dxn.tunnel.client.Constants;
import nl.dxn.tunnel.client.config.IpPortProperties;
import nl.dxn.tunnel.client.config.TcpConfigProperties;
import nl.dxn.tunnel.client.executors.AuthExecutor;
import nl.dxn.tunnel.enums.TransferMessageType;
import nl.dxn.tunnel.po.MessageConfig;
import nl.dxn.tunnel.po.TransferMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 服务器连接处理器
 */
public class ProxyTcpHandler extends SimpleChannelInboundHandler<TransferMessage> {

    private static final Logger logger = LoggerFactory.getLogger(ProxyTcpHandler.class);

    private TcpConfigProperties tcp;

    public ProxyTcpHandler(TcpConfigProperties tcp) {
        this.tcp = tcp;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext context, TransferMessage message) throws Exception {
        try {
            if (message.getType() == TransferMessageType.TYPE_KEEPALIVE) {
                return;
            }

            if (message.getType() == TransferMessageType.TYPE_AUTH) {
                AuthExecutor.execute(context, message);
                return;
            }

            if (message.getType() == TransferMessageType.TYPE_CONNECTED) {
                dealConnect(context, message);
                return;
            }
            if (message.getType() == TransferMessageType.TYPE_DISCONNECTED) {
                dealDisConnect(context, message);
                return;
            }
            if (message.getType() == TransferMessageType.TYPE_DATA) {
                dealTransferData(context, message);
                return;
            }
        } catch (Exception ex) {
            logger.error("", ex);
        }
    }

    /**
     * 断开连接
     *
     * @param ctx
     * @throws Exception
     */

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("tcp隧道客户端-服务端代理连接中断");
        AuthExecutor.executeAgentDisConnect(ctx);
        logger.info("tcp隧道客户端-服务端代理连接，后事处理完毕");
    }

    /**
     * 连接异常
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("",cause);
        ctx.close();
    }

    public void dealConnect(ChannelHandlerContext context, TransferMessage message) throws Exception {
        MessageConfig messageConfig = message.getConfig();
        try {
            String ip = tcp.getDefaultIp();
            int port = tcp.getDefaultPort();

            if (StringUtils.isEmpty(ip) || port == 0) {
                sendNoClintMessage(context, messageConfig);
                return;
            }
            createLocalTcpChannel(context, messageConfig, ip, port);
        }
        catch (Exception exception)
        {
            logger.error("agent客户端连接本地服务失败", exception);
            sendLocalClintConnectFaiMessage(context, messageConfig);
        }
    }

    private void sendNoClintMessage(final ChannelHandlerContext context, final MessageConfig messageConfig) {
        TransferMessage resultMassage = new TransferMessage();
        resultMassage.setConfig(messageConfig);
        resultMassage.setData(EmptyArrays.EMPTY_BYTES);
        resultMassage.setType(TransferMessageType.TYPE_CONNECTED);
        resultMassage.getConfig().setResult("代理客户端没有配置代理地址");
        context.writeAndFlush(resultMassage);
    }

    private void sendLocalClintConnectFaiMessage(final ChannelHandlerContext context, final MessageConfig messageConfig) {
        TransferMessage resultMassage = new TransferMessage();
        resultMassage.setConfig(messageConfig);
        resultMassage.setData(EmptyArrays.EMPTY_BYTES);
        resultMassage.setType(TransferMessageType.TYPE_CONNECTED);
        resultMassage.getConfig().setResult("代理客户端远程地址连接失败");
        context.writeAndFlush(resultMassage);
    }

    private void sendLocalClintConnectMessage(final ChannelHandlerContext context, final MessageConfig messageConfig) {
        TransferMessage resultMassage = new TransferMessage();
        resultMassage.setConfig(messageConfig);
        resultMassage.setData(EmptyArrays.EMPTY_BYTES);
        resultMassage.setType(TransferMessageType.TYPE_CONNECTED);
        resultMassage.getConfig().setResult(null);
        context.writeAndFlush(resultMassage);
    }

    private void createLocalTcpChannel(final ChannelHandlerContext context, final MessageConfig messageConfig, String ip, int port) throws InterruptedException {
        try {
            logger.info("agent客户端连接本地服务{} {} {}", messageConfig.getServerId(),ip,port);
            EventLoopGroup localGroup = new NioEventLoopGroup();
            Bootstrap localBootstrap = new Bootstrap();
            localBootstrap.group(localGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast(new FlushConsolidationHandler(5));
                            channel.pipeline().addLast(new TcpClientHandel(messageConfig));
                        }
                    });

            //启动服务器并绑定一个端口并且同步生成一个 ChannelFuture 对象
            ChannelFuture cf = localBootstrap.connect(ip, port).sync();
            if (cf.isSuccess()) {
                // 本地连接注册代理的隧道
                cf.channel().attr(Constants.PROXY).set(context.channel());
                // 隧道中记录代理的客户端对应关系  (发起方)clientId-->localChannel(本地服务方)
                ConcurrentHashMap<String, Pair<EventLoopGroup, Channel>> clientLocalMap = context.channel().attr(Constants.LOCAL).get();
                if (clientLocalMap == null) {
                    clientLocalMap = new ConcurrentHashMap<>();
                    context.channel().attr(Constants.LOCAL).set(clientLocalMap);
                }
                Pair<EventLoopGroup, Channel> pair = clientLocalMap.get(messageConfig.getChannelId());
                if (pair != null) {
                    try {
                        pair.getRight().close();
                    } catch (Exception ex) {
                        logger.error("", ex);
                    }
                    try {
                        pair.getKey().shutdownGracefully();
                    } catch (Exception ex) {
                        logger.error("", ex);
                    }
                }
                clientLocalMap.put(messageConfig.getChannelId(), Pair.of(localGroup, cf.channel()));
                sendLocalClintConnectMessage(context, messageConfig);
            } else {
                logger.info("agent客户端连接服务端失败,服务端地址IP:{},Http端口:{},原因：{}", ip, port, cf.cause().getMessage());
                sendLocalClintConnectFaiMessage(context, messageConfig);
                localGroup.shutdownGracefully();
            }
        }
        catch (Exception exception)
        {
            logger.error("agent客户端连接服务端失败,服务端地址IP:port"+ip+port, exception);
            sendLocalClintConnectFaiMessage(context, messageConfig);
        }
    }

    public void dealDisConnect(ChannelHandlerContext context, TransferMessage message) {
        try {
            logger.info("proxy {} {} {} 断开连接",message.getConfig().getServerId(),message.getConfig().getChannelId(),context.channel().remoteAddress());
            ConcurrentHashMap<String, Pair<EventLoopGroup, Channel>> localChannelMap = context.channel().attr(Constants.LOCAL).get();
            if (CollectionUtils.isEmpty(localChannelMap)) {
                return;
            }
            // 找到对应的本地连接，断开连接
            Pair<EventLoopGroup, Channel> pair = localChannelMap.get(message.getConfig().getChannelId());
            if (pair != null) {
                localChannelMap.remove(message.getConfig().getChannelId());
                if (pair.getRight() != null) {
                    try {
                        pair.getRight().attr(Constants.PROXY).set(null);
                        pair.getRight().close();
                    } catch (Exception ex) {
                        logger.error("", ex);
                    }
                }
                if (pair.getKey() != null) {
                    try {
                        pair.getKey().shutdownGracefully();
                    } catch (Exception ex) {
                        logger.error("", ex);
                    }
                }
            }
        } catch (Exception ex) {
            logger.error("", ex);
        }
    }

    public void dealTransferData(ChannelHandlerContext context, TransferMessage message) {
        ConcurrentHashMap<String, Pair<EventLoopGroup, Channel>> localChannelMap = context.channel().attr(Constants.LOCAL).get();
        if (CollectionUtils.isEmpty(localChannelMap)) {
            return;
        }
        Pair<EventLoopGroup, Channel> pair = localChannelMap.get(message.getConfig().getChannelId());
        if (pair.getRight() != null) {
            ByteBuf byteBuf = Unpooled.buffer(message.getData().length);
            byteBuf.writeBytes(message.getData());
            ChannelFuture future = pair.getRight().writeAndFlush(byteBuf);
//            if(future.isSuccess())
//            {
//                byteBuf.release();
//            }
        }
    }
}
